package com.ferry.transport.netty.consumer;

import com.ferry.protocol.ProtocolDecoder;
import com.ferry.protocol.ProtocolEncoder;
import com.ferry.transport.api.Connection;
import com.ferry.transport.api.UnResolvedAddress;
import com.ferry.transport.api.channel.WrappedChannelGroup;
import com.ferry.transport.api.config.TcpOption;
import com.ferry.transport.api.domain.Protocol;
import com.ferry.transport.netty.handler.client.ConnectorHandler;
import com.ferry.transport.netty.handler.client.ConnectorIdleStateTrigger;
import com.ferry.transport.netty.handler.client.DefaultConsumerProcessor;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * Created by daiyong
 */
public class DefaultNettyTcpConnector extends NettyTcpConnector{

	private final ConnectorIdleStateTrigger connectorIdleStateTrigger = new ConnectorIdleStateTrigger();


	public DefaultNettyTcpConnector() {
		super();
	}

	public DefaultNettyTcpConnector(Protocol protocol) {
		super(protocol);
	}

	public DefaultNettyTcpConnector(int nWorkers) {
		super(nWorkers);
	}

	@Override
	public void setIoRatio(int workerIoRatio) {

	}

	@Override
	public void doInit() {
		config().setOption(TcpOption.SO_REUSEADDR, true);
		config().setOption(TcpOption.CONNECT_TIMEOUT_MILLIS, (int) TimeUnit.SECONDS.toMillis(3));
		initChannelFactory();
	}

	@Override
	public void setProcessor(ConsumerProcessor processor) {
		//todo
	}

	@Override
	public Connection connect(UnResolvedAddress address, boolean async) {
		setOptions();

		final Bootstrap boot = bootstrap();
		final SocketAddress socketAddress = InetSocketAddress.createUnresolved(address.getIp(), address.getPort());
		final WrappedChannelGroup group = group(address);

		// 重连
		final ConnectionKeeper connectionKeeper = new ConnectionKeeper(boot, timer, socketAddress, group) {

			@Override
			public ChannelHandler[] handlers() {
				ChannelHandler[] channelHandlers = {
						this,
						new com.ferry.transport.netty.handler.client.IdleStateChecker(timer, 0, 2, 0),
						connectorIdleStateTrigger,
						new ProtocolDecoder(),
						new ProtocolEncoder(),
						new ConnectorHandler(new DefaultConsumerProcessor())
				};
				return channelHandlers;
			}
		};

		ChannelFuture future;
		try {
			synchronized (boot) {
				boot.handler(new ChannelInitializer<Channel>() {

					@Override
					protected void initChannel(Channel ch) throws Exception {
						ch.pipeline().addLast(connectionKeeper.handlers());
					}
				});

				future = boot.connect(socketAddress);
			}

			if (!async) {
				future.sync();
			}
		} catch (Throwable t) {
			throw new RuntimeException("Connects to [" + address + "] fails", t);
		}

		return new NettyConnection(address, future) {
			@Override
			public void setReconnect(boolean reconnect) {
				if (reconnect) {
					connectionKeeper.start();
				} else {
					connectionKeeper.stop();
				}
			}
		};
	}
}
